package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * @ClassName: SessSourBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:10 
 *
 */
public class SdsSessStat5MinBolt extends AbstractBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = -5475897483936259084L;



	HashMap<String,Long> sdsSessStat5MinMap = new HashMap<String,Long>();
	HashMap<String,Long> sdsSessStat5MinMap_Packet = new HashMap<String,Long>();

//	HashSet<String> IPSet = new HashSet<String>();
//	Slave s = null;

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {

			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("SdsSessStat5Bolt Exception",e);
		}	

	}

	private void countInLocal(Tuple tuple) {

		String dip = tuple.getStringByField("IP");

		String type = tuple.getStringByField("TYPE");

		long nOCTETS = tuple.getLongByField("OCTETS");
		
		long numP = tuple.getLongByField("PACKET");


		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyyMMddHH");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH_ = sdfh_.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		String sMin5 = String.format("%02d", nMin5);
		String dateStr5 = dateStrH_+sMin5;



		//###5分钟
		String rowkey5Min = String.format("%s|%s|%s", dip, dateStr5 , type);
		if(sdsSessStat5MinMap.containsKey(rowkey5Min))
		{
			long dipPV = 0;
				
			dipPV = sdsSessStat5MinMap.get(rowkey5Min) + nOCTETS;
			sdsSessStat5MinMap.put(rowkey5Min, dipPV);
			
			long pNum = 0 ;
			pNum = sdsSessStat5MinMap.get(rowkey5Min) + numP;
			sdsSessStat5MinMap_Packet.put(rowkey5Min, pNum);
			
		}
		else
		{
				
				sdsSessStat5MinMap.put(rowkey5Min, nOCTETS);
				sdsSessStat5MinMap_Packet.put(rowkey5Min, numP);
		}

	}

	private void emitCountingData() {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);

//		if(s.GetIPFilterForLog() != null && s.GetIPFilterForLog().size() > 0) {
//			IPSet = s.GetIPFilterForLog();
//		}
		
		for(String key : sdsSessStat5MinMap.keySet()) {
			String[] split = key.split("\\|",-1);
			String dip = split[0];
			String dateStr = split[1];
			String type = split[2];
			long typePV = sdsSessStat5MinMap.get(key);

//			if(IPSet.size() == 0 || IPSet.contains(dip)) {

				String rowkey = String.format("%s_%s", dip, dateStr);
				String tbName = ApolloProperties.SDS_SESSION_STAT_5MIN;
				hbaseInstance.Add(tbName, rowkey, "cf", type, typePV);
//				System.out.println(">>>>>>>>>>SdsSessStat5Min<<<<<<<<<<<< key =" + rowkey + "; pv =" + typePV + "; date= " + dateStr + "; type= " + type);
				
				if(sdsSessStat5MinMap_Packet.get(key)!=0) {
					hbaseInstance.Add(tbName, rowkey, "cf", type+"NUM", sdsSessStat5MinMap_Packet.get(key));
					
				}
//				hbaseInstance.incerment(tbName, rowkey, "cf:"+type, (long)typePV);
//			}
		}

		sdsSessStat5MinMap.clear();

	}


	

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
//		s = new Slave(zooserver);
//		IPSet = s.GetIPFilterForLog();
	}


}
